package com.digiwin.muke;

import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.configuration.Configuration;

import java.io.File;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author yanggld
 * @Date 2021/04/25-18:01
 * flink分布式缓存
 */
public class DistributedCacheApp {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // register a file from HDFS
        env.registerCachedFile("hdfs://c1:9000/hello.txt", "hello.txt");

        // 存放分布式缓存的内容，。，。。
        List<String> fileContext = new ArrayList<>();

        DataSource<String> data = env.fromElements("hello hihao","hello world1111");
        data.map(new RichMapFunction<String, String>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                File myFile = getRuntimeContext().getDistributedCache().getFile("hello.txt");
                List<String> lines = FileUtils.readLines(myFile);
                for (String line : lines) {
                    fileContext.add(line);
                    System.out.println("-----------"+line+"---------");
                }
            }

            @Override
            public String map(String value) throws Exception {
                if (fileContext.contains(value)) {
                    value = value+"存在分布式缓存中";
                } else {
                    value = value+"不存在分布式缓存中";
                }
                return value;
            }

        }).print();
    }
}
